package org.data;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.yarn.webapp.view.HtmlPage;
import org.sort.Floyd;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;

public class Clean5 {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        //1 获取job对象
        Configuration conf = new Configuration();
        conf.set("mapred.textoutputformat.separator",",");
        conf.set("mapred.textoutputformat.ignoreseparator","true");
        Job job = Job.getInstance(conf);

        //2 关联Driver类
        job.setJarByClass(Clean5.class);

        //3 关联Mapper和Reducer类
        job.setMapperClass(Clean5Mapper.class);
        job.setReducerClass(Clean5Reducer.class);

        //4 设置Map的输入输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //5	设置最终的输入输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //6	设置输入输出路径
        FileInputFormat.setInputPaths(job,new Path("D:\\data\\out4\\part-r-00000"));
        FileOutputFormat.setOutputPath(job,new Path("D:\\data\\out5"));

        //7 提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result?0:1);

    }

}

class Clean5Mapper extends Mapper<Object, Text,Text,Text> {
    private Text outK = new Text();

    private Text outV = new Text();
    HashMap<String, Integer> _hashMap11 = new HashMap<>();
    HashMap<String, String> _hashMap33 = new HashMap<>();
    int x=0,y=0;
    int count1=0;
    int door1=11645;

    protected void map(Object key,Text value,Context context) throws IOException,InterruptedException{

        String[] str = value.toString().split(",");
        String s1 = str[0];
        String s2 = str[1];
        String s3 = str[2];

        //第一个哈希表
        HashMap<String, Integer> hashMap11 = new HashMap<>();
        if (!_hashMap11.isEmpty()){
            if(!_hashMap11.containsKey(s1)){
                hashMap11.put(s1,x);
                _hashMap11.putAll(hashMap11);
                x++;
            }
        }else {
            hashMap11.put(s1,x);
            _hashMap11.putAll(hashMap11);
            x++;
        }
        if (!_hashMap11.isEmpty()){
            if(!_hashMap11.containsKey(s2)){
                hashMap11.put(s2,x);
                _hashMap11.putAll(hashMap11);
                x++;
            }
        }else {
            hashMap11.put(s2,x);
            _hashMap11.putAll(hashMap11);
            x++;
        }

        //第二个哈希表
        HashMap<String, String> hashMap33 = new HashMap<>();
        if (!_hashMap33.isEmpty()){
            if(!_hashMap33.containsKey(s1+","+s2)){
                hashMap33.put(s1+","+s2,s3);
                _hashMap33.putAll(hashMap33);
            }
        }else {
            hashMap33.put(s1+","+s2,s3);
            _hashMap33.putAll(hashMap33);
        }

        count1++;
        if (count1 == door1){

            String realk1=null;
            String realk2=null;
            String realk=null;
            String realv=null;

            int[][] a=SortFloyd(_hashMap11,_hashMap33);
            System.out.printf("floyd: \n");

            for (int i = 0; i < a.length; i++) {
                for (int j = 0; j < a.length; j++) {
                    //
                    for (String key111 : _hashMap11.keySet()) {
                        if (_hashMap11.get(key111).equals(i)) {
                            realk1 = key111;
                        }
                    }
                    for (String key111 : _hashMap11.keySet()) {
                        if (_hashMap11.get(key111).equals(j)) {
                            realk2 = key111;
                        }
                    }
                    Double d1= Double.valueOf(a[i][j]);
                    Double d = d1/=100;

                    realk = realk1+","+realk2;
                    realv = String.valueOf(d);
                    if (realv.equals("2.147483647E7")){
                        realv = "不可达";
                    }

                    outK.set(realk);
                    outV.set(realv);
                    context.write(outK,outV);
                }//END FOR
            }//END FOR
        }//END IF

    }

    private int[][] SortFloyd(HashMap<String, Integer> hashMap11, HashMap<String, String> hashMap33) {
        int INF = Integer.MAX_VALUE;
        int[][] Matrix=new int[hashMap11.size()][hashMap11.size()];
        char[] Nodes=new char[hashMap11.size()];

        for (int i=0; i<Matrix.length;i++){
            for (int j=0; j<Matrix[i].length;j++){
                Matrix[i][j]=INF;
            }
        }

        for (int i=0;i<hashMap11.size();i++){
            Nodes[i]= (char) i;
        }

        Iterator<String> iterator1 = _hashMap33.keySet().iterator();

        //
        while (iterator1.hasNext()){
            String key3 = iterator1.next();//key
            int val3 = (int) (Double.valueOf(_hashMap33.get(key3))*100);//value

            System.out.println("key3"+key3);



            String[] str = key3.split(",");
            int x = hashMap11.get(str[0]);
            int y = hashMap11.get(str[1]);

            System.out.println(val3);
            System.out.println("坐标:"+x+"---"+y);
            Matrix[x][y]= val3;

        }

        for (int i=0; i<Matrix.length;i++){
            for (int j=0; j<Matrix[i].length;j++){
               try {
                   System.out.println(Matrix[i][j]);
               }catch (Exception e){
                   e.printStackTrace();
               }
            }
        }

        int[] dist = new int[Nodes.length];
        Floyd floyd = new Floyd(Nodes, Matrix);

        return floyd.floyd();
    }
}

class Clean5Reducer extends Reducer<Text,Text,Text, Text> {
    public void reduce(Text key, Iterable<Text> values,Context context) throws IOException,InterruptedException {


        Text content = new Text();
        for (Text value:values){
            content.set(value);
        }
        context.write(key, content);
    }
}